package com.yk.spark.model.submit;

import com.yk.spark.model.AppModel;
import com.yk.spark.model.FileModel;
import com.yk.spark.service.file.api.IFileService;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;

import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * Created by Limit on 2017/8/24.
 */
public class SubmitOptions {

    private String[] appArgs;
    private String appResource;
    private Map<String,String> environmentVariables;
    private Map<String,String> sparkProperties;
    private String mainClass;
    private IFileService fileService;

    public SubmitOptions(AppModel model, IFileService fileService) {

        this.fileService=fileService;
        try {
            JSONObject parameters=new JSONObject(model.getAppParameters());
            this.appArgs=this.buildAppArgs(model.getAppName(),parameters.getString("master"),new JSONArray(model.getAppInput()),model.getAppOutput());
            this.environmentVariables=this.buildEnvironmentVariables(parameters.getJSONObject("environmentVariables"));
            this.sparkProperties=this.buildSparkProperties(parameters.getJSONObject("sparkProperties"));
            this.appResource=parameters.getString("appResource");
            this.mainClass=parameters.getString("mainClass");
        } catch (JSONException e) {
            e.printStackTrace();
        }
    }

    private  String[] buildAppArgs(String appName,String master,JSONArray input,String output){
        StringBuilder in= new StringBuilder();
        for (int i = 0; i <input.length() ; i++) {
            try {
                FileModel model=this.fileService.getFileModel(input.getString(i));
                in.append(model.getFilePath()).append(model.getFileName()).append(" ");
            } catch (JSONException e) {
                e.printStackTrace();
            }
        }

        return new String[]{
                "-in",in.toString(),
                "-out",output,
                "-appname",appName,
                "-master",master
        };
    }

    private  Map<String,String> buildEnvironmentVariables(JSONObject environmentVariables){
        return new HashMap<String, String>() {{
            Iterator keys = environmentVariables.keys();
            while (keys.hasNext()) {
                String key = (String) keys.next();
                try {
                    put(key, environmentVariables.getString(key));
                } catch (JSONException e) {
                    e.printStackTrace();
                }
            }
        }};
    }

    private Map<String,String> buildSparkProperties(JSONObject sparkProperties){

        StringBuilder jarStr = new StringBuilder();
        try {
            JSONArray jars=sparkProperties.getJSONArray("spark.jars");
            for (int i = 0; i <jars.length() ; i++) {
                jarStr.append(jars.getString(i)).append(" ");
            }
        } catch (JSONException e) {
            e.printStackTrace();
        }

        return new HashMap<String, String>() {{
            Iterator keys = sparkProperties.keys();
            while (keys.hasNext()) {
                String key = (String) keys.next();
                try {
                    if (key.equals("spark.jars")) {
                        put(key, jarStr.toString());
                    } else {
                        put(key, sparkProperties.getString(key));
                    }
                } catch (JSONException e) {
                    e.printStackTrace();
                }
            }
        }};
    }

    public String[] getAppArgs() {
        return appArgs;
    }

    public void setAppArgs(String[] appArgs) {
        this.appArgs = appArgs;
    }

    public String getAppResource() {
        return appResource;
    }

    public void setAppResource(String appResource) {
        this.appResource = appResource;
    }

    public Map<String, String> getEnvironmentVariables() {
        return environmentVariables;
    }

    public void setEnvironmentVariables(Map<String, String> environmentVariables) {
        this.environmentVariables = environmentVariables;
    }

    public Map<String, String> getSparkProperties() {
        return sparkProperties;
    }

    public void setSparkProperties(Map<String, String> sparkProperties) {
        this.sparkProperties = sparkProperties;
    }

    public String getMainClass() {
        return mainClass;
    }

    public void setMainClass(String mainClass) {
        this.mainClass = mainClass;
    }
}
